package com.monkeyboy.demo.guigu.partition;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**
 * @Author Gavin
 * @date 2021.03.12 13:05
 */
public class MyProducerPartition {
    public final static String TOPIC_NAME = "first";

    public static void main(String[] args) throws Exception {
        //带回调的生产者
        producerCallBack();

    }
    /*
    带回调的生产者
    */
    public static void producerCallBack() throws Exception {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.150:9092");
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        //消息重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG, "0");

        //数据达到16k会发送
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        //当时间到达1ms的时候也会发送和上面两个结合起来使用
        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");
        //数据缓冲区，32M
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");

        //序列化设置
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        //指定自定义的分区类
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.monkeyboy.demo.guigu.partition.MyPartitioner");

        //Producer的主对象
        Producer<String, String> producer = new KafkaProducer<>(properties);
        //组织消息对象：ProducerRecord
        for (int i = 0; i < 10; i++) {
            //ProducerRecord有很多构造函数，可以指定要发送的partition，
            //如果指定key就按照key的hash进行指定分区
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, "key:" + i, "value:" + i);
            /**
             * 成功的回调，如果成功，返回recordMetadata，否则返回异常
             */
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if(e==null){
                        System.out.println("【分区】"+recordMetadata.partition()+"   【offset】 "+recordMetadata.offset());
                    }else{
                        e.printStackTrace();
                    }
                }
            });
        }
        Thread.sleep(100);
        //如果这里关闭的时候上面执行的时间没有达到1ms，那么消息会被清空，然后消息也不会进行发送
        producer.close();
    }
}
